package com.dark.trident;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by tengxue on 16-6-30.
 */
public class HourAssignment extends  BaseFunction{

    private static final long serialVersionUID=1L;

    private static final Logger LOG= LoggerFactory.getLogger(HourAssignment.class);


    public void execute(TridentTuple tuple, TridentCollector collector) {
        DiagnosisEvent diagnosisEvent=(DiagnosisEvent)tuple.getValue(0);
        String city=(String)tuple.getValue(1);
        long timestamp=diagnosisEvent.getTime();
        long hourSinceEpoch=timestamp/1000/60/60;
        LOG.info("Key = [" + city+" : "+hourSinceEpoch+" ]");
        String key=city+":"+diagnosisEvent.getDiagnosisCode()+":"+hourSinceEpoch;
        List<Object> values=new ArrayList<Object>();
        values.add(hourSinceEpoch);
        values.add(key);
        collector.emit(values);

    }
}
